package org.example.flinktest.apidemo;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Random;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.example.flinktest.bean.SensorReading;


/**
 * @author shihb
 * @date 2019/12/6 18:45
 */
public class SourceDemo {


  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    List sensorMsg=new ArrayList<SensorReading>();
    sensorMsg.add(new SensorReading("sensor_1",1575631190L,28.333644));
    sensorMsg.add(new SensorReading("sensor_2",1575633190L,31.644865));
    sensorMsg.add(new SensorReading("sensor_4",1575634190L,19.328949));

    //自定义从集合中获取数据
    DataStream <SensorReading> stream1=env.fromCollection(sensorMsg);

    //从文件中读取数据
    DataStream <String> stream2=env.readTextFile("D:\\SHBData\\IDEAProjects\\fink-parent\\flink-quickstart-java\\src\\main\\resources\\sensor.txt");

    //从kafka读取
    Properties properties=new Properties();
    properties.setProperty("bootstrap.servers","192.168.7.201:9092,192.168.7.202:9092,192.168.7.203:9092");
    properties.setProperty("group.id", "consumer-group");
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.setProperty("auto.offset.reset","latest");
    DataStream <String> stream3=env.addSource(new FlinkKafkaConsumer010<String>("test",new SimpleStringSchema(),properties));

    //自定义source
    DataStream <SensorReading> stream4=env.addSource(new SensorSource());

    stream4.print("stream4").setParallelism(1);


    env.execute("source demo");

  }
}
class SensorSource  implements SourceFunction<SensorReading>{
  //定义一个flag表示数据是否正常
  private  boolean running=true;

  //正常生成数据
  @Override
  public void run(SourceContext<SensorReading> sourceContext) throws Exception {
    Random random=new Random();
    //定义一组传感器
    List<Tuple2<String,Double>> list=new ArrayList<Tuple2<String,Double>>();
    for (int i = 1; i <=10 ; i++) {
      list.add(new Tuple2<>("sensor_"+i,60+random.nextGaussian()*20));
    }


    while(running){

      Long curTime=System.currentTimeMillis();
      for (int i = 0; i <list.size() ; i++) {
        Tuple2<String, Double> t=list.get(i);
        list.set(i,new Tuple2<String, Double>(t.f0,t.f1+random.nextGaussian()));
        sourceContext.collect(new SensorReading(t.f0,curTime,t.f1));
      }
      System.out.println(list);
      Thread.sleep(10000);
    }
  }
  //取消数据源的生成
  @Override
  public void cancel() {
    running=false;
  }
}
